package com.example.flinkcourse.lesson4.watermark;

import com.example.flinkcourse.lesson4.model.Event;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 自定义水印生成器
 * 用于生成事件时间水印
 */
public class CustomWatermarkGenerator implements WatermarkGenerator<Event> {
    private static final Logger LOG = LoggerFactory.getLogger(CustomWatermarkGenerator.class);
    
    private long maxTimestamp;
    private final long maxOutOfOrderness;
    
    /**
     * 构造函数
     * @param maxOutOfOrderness 最大乱序时间（毫秒）
     */
    public CustomWatermarkGenerator(long maxOutOfOrderness) {
        this.maxOutOfOrderness = maxOutOfOrderness;
        this.maxTimestamp = Long.MIN_VALUE + maxOutOfOrderness;
    }
    
    @Override
    public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {
        LOG.debug("Processing event timestamp: {}", eventTimestamp);
        maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
    }
    
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        long watermark = maxTimestamp - maxOutOfOrderness;
        LOG.debug("Emitting watermark: {}", watermark);
        output.emitWatermark(new Watermark(watermark));
    }
    
    /**
     * 创建水印生成器
     * @param maxOutOfOrderness 最大乱序时间（毫秒）
     * @return 水印生成器
     */
    public static WatermarkGenerator<Event> create(long maxOutOfOrderness) {
        return new CustomWatermarkGenerator(maxOutOfOrderness);
    }
    
    /**
     * 创建默认水印生成器
     * @return 水印生成器
     */
    public static WatermarkGenerator<Event> createDefault() {
        return new CustomWatermarkGenerator(5000); // 默认5秒乱序
    }
} 